package per.chao.mqtt.domain.repo;

import com.google.common.collect.Lists;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.configurationprocessor.json.JSONObject;
import org.springframework.stereotype.Component;
import per.chao.mqtt.domain.interfaces.repo.BaseSubscribeRepo;
import per.chao.mqtt.domain.repo.subscribe.SubscribeNotWildcardTopicRepo;
import per.chao.mqtt.domain.repo.subscribe.SubscribeWildcardTopicRepo;
import per.chao.mqtt.domain.shared.store.SubscribeStore;
import per.chao.mqtt.domain.util.TaskCircle;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Map;
import java.util.TimerTask;
import java.util.concurrent.TimeUnit;

/**
 * Description:
 *
 * @author W.Chao
 * @date 2021/6/20 14:32
 **/
@Component
@Slf4j
public class SubscribeMessageRepo extends BaseSubscribeRepo<SubscribeStore> {
	private static final String SKIP_ONE_LEVEL = "+";
	private static final String ONLY_IN_LAST_AND_MEANING_ANY_WORDS = "#";

	@Autowired
	private SubscribeNotWildcardTopicRepo notWildcardRepo;

	@Autowired
	private SubscribeWildcardTopicRepo wildcardRepo;

	@PostConstruct
	public void construct() {

		TaskCircle.addTask(new TimerTask() {
			@SneakyThrows
			@Override
			public void run() {
				Map<String, Map<String, SubscribeStore>> notWildcard = notWildcardRepo.all();
				Map<String, Map<String, SubscribeStore>> wildcard = wildcardRepo.all();
				JSONObject repo = new JSONObject();
				repo.put("notWildcard", notWildcard);
				repo.put("wildcard", wildcard);
				log.info("current subscribe: {}", repo);
			}
		}, TimeUnit.SECONDS.toMillis(1), TimeUnit.SECONDS.toMillis(10));
	}

	@Override
	public void put(String topic, String clientId, SubscribeStore subscribeStore) {
		if (containsWildcard(topic)) {
			wildcardRepo.put(topic, clientId, subscribeStore);
		} else {
			notWildcardRepo.put(topic, clientId, subscribeStore);
		}
	}

	@Override
	public SubscribeStore get(String topic, String clientId) {
		List<SubscribeStore> stores = getByTopic(topic);
		if (stores != null) {
			return stores.stream()
					.filter(store -> store.getClientId().equals(clientId))
					.findFirst()
					.orElse(null);
		}
		return null;
	}

	public List<SubscribeStore> getByTopic(String topic) {
		List<SubscribeStore> stores = Lists.newArrayList();
		List<SubscribeStore> notWildcardSubAll = notWildcardRepo.all(topic);
		if (notWildcardSubAll != null) {
			stores.addAll(notWildcardSubAll);
		}

		wildcardRepo.all().forEach((topicFilter, map) -> {
			if (topic.split("/").length >= topicFilter.split("/").length) {
				String[] originalTopic = topic.split("/");
				String[] filterTopic = topicFilter.split("/");
				StringBuilder newTopicFilter = new StringBuilder();
				for (int i = 0; i < filterTopic.length; i++) {
					String fragment = filterTopic[i];
					if (SKIP_ONE_LEVEL.equals(fragment)){
						// 忽略该层的topic片段值
						newTopicFilter.append(SKIP_ONE_LEVEL).append("/");
					} else if (ONLY_IN_LAST_AND_MEANING_ANY_WORDS.equals(fragment)) {
						// 忽略该层的topic片段值
						newTopicFilter.append(ONLY_IN_LAST_AND_MEANING_ANY_WORDS).append("/");
					} else {
						// 非通配符部分
						newTopicFilter.append(originalTopic[i]).append("/");
					}
				}
				// 去除多余的最后一个/
				String finalNewTopicFilter = newTopicFilter.substring(0, newTopicFilter.length() - 1);
				if (topicFilter.equals(finalNewTopicFilter)) {
					stores.addAll(map.values());
				}
			}
		});

		return stores;
	}

	@Override
	public void remove(String topic, String clientId) {
		if (containsWildcard(topic)) {
			wildcardRepo.remove(topic, clientId);
		} else {
			notWildcardRepo.remove(topic, clientId);
		}
	}

	@Override
	@Deprecated
	public boolean containsKey(String topic, String clientId) {
		return false;
	}

	@Override
	public void remove(String clientId) {
		wildcardRepo.remove(clientId);
		notWildcardRepo.remove(clientId);
	}

	/**
	 * topic是否包含通配符
	 *
	 * @param topic
	 * @return
	 */
	private boolean containsWildcard(String topic) {
		return topic.contains(SKIP_ONE_LEVEL) || topic.contains(ONLY_IN_LAST_AND_MEANING_ANY_WORDS);
	}
}
